package com.typhoon.rpc.registry.zk;

import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.typhoon.rpc.common.util.CollectionUtil;
import com.typhoon.rpc.registry.ServiceDiscovery;
import com.typhoon.rpc.registry.constant.Constant;

public class ZkServiceDiscovery implements ServiceDiscovery {


    private static final Logger LOGGER = LoggerFactory.getLogger(ZkServiceDiscovery.class);

    private String zkAddress;

    public ZkServiceDiscovery(String zkAddress) {
        this.zkAddress = zkAddress;
    }

    @Override
    public String discover(String name) {
        // 创建 ZooKeeper 客户端
        ZkClient zkClient = new ZkClient(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT);
        //①先创建zk连接,类似jdbc
        LOGGER.debug("connect zookeeper");
        try {
            // 获取 service 节点
            String servicePath = Constant.ZK_REGISTRY_PATH + "/" + name;
            if (!zkClient.exists(servicePath)) {//如果服务节点不存在(尚未有服务注册),抛出运行异常
                throw new RuntimeException(String.format("can not find any service node on path: %s", servicePath));
            }
            List<String> addressList = zkClient.getChildren(servicePath);//获取服务节点的所有子节点
            if (CollectionUtil.isEmpty(addressList)) {//没有可用的服务实例(尚未有服务注册,或者服务宕机不可用)
                throw new RuntimeException(String.format("can not find any address node on path: %s", servicePath));
            }
            // 获取 address 节点
            String address;
            int size = addressList.size();
            if (size == 1) {//如果只有一个服务,获取该地址
                // 若只有一个地址，则获取该地址
                address = addressList.get(0);
                LOGGER.debug("get only address node: {}", address);
            } else {//如果有多个服务实例,随机获取,(根据有种负载策略获取)
                // 若存在多个地址，则随机获取一个地址
                address = addressList.get(ThreadLocalRandom.current().nextInt(size));
                LOGGER.debug("get random address node: {}", address);
            }
            // 获取 address 节点的值
            String addressPath = servicePath + "/" + address;//拼装到具体的服务实例节点
            return zkClient.readData(addressPath);//读取节点并返回
        } finally {
            zkClient.close();
        }
    }

}
